package jmars.mobile.worker;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import jmars.mobile.common.Investigator;
import jmars.mobile.datainvestigation.Summary;

/**
 * select [group columns], count, sum(target column1) [, sum(target column2)...] 
 * from data group by [group columns] 
 * order by count desc
 * 
 * @author Jan
 *
 */
public class GroupSumEach implements Investigator {

	@Override
	public Long compute(File[] files, FileWriter writer) throws IOException {
		Long total = 0L;
		FileInputStream fis;
		BufferedReader reader;
		String row;
		String[] cols;
		Map<String,Integer> distinct=new HashMap<String,Integer>();
		Map<String,Double> order=new HashMap<String,Double>();
		Map<String,Map<Integer,Double>> sumMap=new HashMap<String,Map<Integer,Double>>();
		Map<Integer,Double> sumEach;
		String groupKey;
		DecimalFormat df = new DecimalFormat();
		df.applyPattern("0");
		Double tempD;
		for(File inputFile:files){
			fis = new FileInputStream(inputFile);
			reader = new BufferedReader(new InputStreamReader(fis));
			while((row=reader.readLine())!=null){
				cols=row.split(Summary.columnSeperator);
				if(Summary.filterRow(cols,true,true)) continue;
				groupKey="";
				for(Integer idx:Summary.groupColumnIdx){
					if(idx<cols.length) {
						groupKey=groupKey+"$_$"+cols[idx]+"$_$";
					}
					else groupKey=groupKey+"$_$$_$";
				}
				if(!sumMap.containsKey(groupKey)){
					distinct.put(groupKey,1);
					order.put(groupKey,0D);
					sumEach = new HashMap<Integer,Double>();
					sumMap.put(groupKey, sumEach);
				}else{
					sumEach = sumMap.get(groupKey);
					distinct.put(groupKey,distinct.get(groupKey)+1);
				}
				for(Integer idx:Summary.targetColumnIdx){
					tempD=Double.parseDouble(cols[idx]);
					order.put(groupKey, order.get(groupKey)+tempD);
					if(!sumEach.containsKey(idx)){
						sumEach.put(idx, tempD);
					}else{
						sumEach.put(idx, sumEach.get(idx)+tempD);
					}
				}

				
				total++;
			}
			reader.close();
			fis.close();
			System.out.println(inputFile.getName()+" finished at "+total);
		}
		writer.write("%"+Summary.groupColumns+":"+df.format(distinct.size())+"\r\n");
		writer.write("%total:"+df.format(total)+"\r\n");
		if(distinct.size()<=1000000){
			ArrayList<Entry<String,Double>> l = new ArrayList<Entry<String,Double>>(order.entrySet());    
	        
	        Collections.sort(l, new Comparator<Map.Entry<String, Double>>() {    
	            public int compare(Map.Entry<String, Double> o1, Map.Entry<String, Double> o2) {    
	                return (int) (o2.getValue() - o1.getValue());    
	            }    
	        });

			writer.write("%*****************details******************\r\n%");
			for(Integer idx:Summary.groupColumnIdx){
				writer.write(Summary.columns[idx]+",");
			}
			writer.write("count");
			for(Integer idx:Summary.targetColumnIdx){
				writer.write(",sum("+Summary.columns[idx]+")");
			}
			writer.write("\r\n");
	        for(Entry<String,Double> e : l) {
	            writer.write(e.getKey().replaceAll("\\$_\\$\\$_\\$", ",").replaceAll("\\$_\\$","")+","+distinct.get(e.getKey()));
	            for(Integer idx:Summary.targetColumnIdx){
		            writer.write(","+df.format(sumMap.get(e.getKey()).get(idx)));
	            }
	            writer.write("\r\n");
	        }
	        
		}
		return total;
	}

}
